package tang.client;

import tang.common.*;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.LinkedList;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * @author Tang
 * @classname ClientCnxnSocketNIO
 * @description [  ]
 * @date 2021/2/23 21:14
 */
public class ClientCnxnSocketNIO {

    private final Selector selector = Selector.open();
    /**
     * 改变客户端感兴趣的监听集合
     */
    private SelectionKey selectionKey;

    public ClientCnxn clientCnxn;

    public ClientCnxn.SendThread sendThread;

    /**
     * 先是长度的容量
     */
    private ByteBuffer lenBuffer = ByteBuffer.allocate(4);

    private ByteBuffer incomingBuffer = lenBuffer;

    public ClientCnxnSocketNIO() throws IOException {
    }

    boolean isConnected() {
        return selectionKey != null;
    }

    public void connect(InetSocketAddress address) throws IOException {
        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.configureBlocking(false);
        socketChannel.socket().setTcpNoDelay(true);
        socketChannel.socket().setSoLinger(false, -1);
        selectionKey = socketChannel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ);
        boolean immediateConnect = socketChannel.connect(address);

        while (!socketChannel.finishConnect()) {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        clientCnxn.setState(Zookeeper.States.CONNECTED);
        // 构造一个连接请求，投进outgoingQueue
        Packet packet = new Packet();
        packet.setType(OpCode.Connect);
        packet.request = new ConnectRequest();
        packet.response = new ConnectResponse();
        clientCnxn.getOutgoingQueue()
                .addFirst(packet);
    }

    public void doTransport(LinkedList<Packet> pendingQueue, LinkedList<Packet> outgoingQueue,
                            ClientCnxn clientCnxn) throws IOException {
        selector.select(1000);
        Set<SelectionKey> selectionKeys = selector.selectedKeys();
        selectionKeys.forEach((selectionKey1 -> {
            try {
                if ((selectionKey1.readyOps() & SelectionKey.OP_CONNECT) != 0) {
                    SocketChannel channel = (SocketChannel) selectionKey1.channel();
                    if (channel.finishConnect()) {
                        System.out.println("客户端与服务器连接建立success....");
                    } else {
                        System.out.println("客户端与服务器连接建立fail....");
                    }
                    // 读 | 写 操作
                } else if ((selectionKey1.readyOps() & (SelectionKey.OP_READ | SelectionKey.OP_WRITE)) != 0) {
                    doIo(pendingQueue, outgoingQueue, clientCnxn);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }));

        // 检测outgoingQueue是否为空，不为空的话，开启写操作
        if (this.clientCnxn.getState().isConnected()) {
            if (!outgoingQueue.isEmpty()) {
                enableWrite();
            }
        }
        selectionKeys.clear();
    }

    private void doIo(LinkedList<Packet> pendingQueue, LinkedList<Packet> outgoingQueue, ClientCnxn clientCnxn) throws Exception {
        SocketChannel channel = (SocketChannel) selectionKey.channel();
        // 读操作
        if (selectionKey.isReadable()) {
            channel.read(incomingBuffer);
            // 第一次肯定为false
            if (!incomingBuffer.hasRemaining()) {
                incomingBuffer.flip();
                // 第一次读取数据包的长度
                if (incomingBuffer == lenBuffer) {
                    readLength();
                } else {
                    // 读数据 唤醒线程
                    sendThread.readResponse(incomingBuffer);
                    incomingBuffer.clear();
                    lenBuffer.clear();
                    incomingBuffer = lenBuffer;
                }
            }
        }
        // 写操作
        if (selectionKey.isWritable()) {
            if (!outgoingQueue.isEmpty()) {
                Packet packet = outgoingQueue.removeFirst();
                System.out.println("从outgoingQueue中取出元素发送给server");
                // 加入到pendingQueue
                pendingQueue.add(packet);
                // 构建数据
                packet.createData();
                // 发送给服务器
                channel.write(packet.getDataBuffer());
                if (outgoingQueue.isEmpty()) {
                    // 开启读操作
                    enableRead();
                }
            }
        }
    }

    private void readLength() {
        // 读出数据长度
        int len = incomingBuffer.getInt();
        incomingBuffer = ByteBuffer.allocate(len);
    }

    void enableWrite() {
        int i = selectionKey.interestOps();
        if ((i & SelectionKey.OP_WRITE) == 0) {
            selectionKey.interestOps(i | SelectionKey.OP_WRITE);
        }
    }

    void enableRead() {
        int i = selectionKey.interestOps();
        if ((i & SelectionKey.OP_READ) == 0) {
            selectionKey.interestOps(i | SelectionKey.OP_READ);
        }
    }
}